package org.es.io;

/**
 * Copyright (C) 2012 Eduard Sedov
 *
 * Licensed under the Apache License, Version 2.0 (the "License"); you may not
 * use this file except in compliance with the License. You may obtain a copy of
 * the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
 * License for the specific language governing permissions and limitations under
 * the License.
 */
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;

/**
 * A page has following areas:<br>
 *
 * 1. header - has following structure:<br>
 * - version (1 byte) currently contains 0 value<br>
 * - free size (2 bytes) amount of free space on the page. (It may be greater
 * than upper - lower after removing of items.)<br>
 * - lower (2 bytes) offset to start of free space<br>
 * - upper (2 bytes) offset to end of free space<br>
 * - tailOffset (2 bytes) offset to start of tail of page<br>
 * 2. array - 4 bytes per pairs (offset/length) pointing to the actual item<br>
 * 3. free space - unallocated space of the page<br>
 * 4. items - items themself. This area starts with tailOffset and grows
 * backward.<br>
 * 5. tail - not used. reserved for expansion<br>
 *
 *
 * @author Eduard Sedov
 */
public class Page {

    public static final int HEADER_SIZE = 9;
    public static final int ARRAY_CELL_SIZE = 4;
    protected static final int VERSION = 0;
    protected static final int FREE_SIZE = 1;
    protected static final int LOWER = 3;
    protected static final int UPPER = 5;
    protected static final int TAIL_OFFSET = 7;
    protected ByteBuffer buffer;
    protected int tailOffset;

    public Page(ByteBuffer buffer) {
        this.buffer = buffer;
        tailOffset = 0x0000FFFF & buffer.getShort(TAIL_OFFSET);
        check();
    }

    protected Page(ByteBuffer buffer, int tailSize) {
        this.buffer = initEmptyPage(buffer, tailSize);
        tailOffset = 0x0000FFFF & buffer.getShort(TAIL_OFFSET);
        check();
    }

    public ByteBuffer buffer() {
        return buffer;
    }

    public static Page get(ByteBuffer buffer) {
        return new Page(buffer);
    }

    public static Page create(ByteBuffer buffer) {
        // Defaultly we dont reserve any bytes for tail area.
        return create(buffer, 0);
    }

    public static Page create(ByteBuffer buffer, int tailSize) {
        return new Page(buffer, tailSize);
    }

    public static ByteBuffer initEmptyPage(ByteBuffer buffer, int tailSize) {
        buffer.put(VERSION, (byte) 0);
        int upper = buffer.limit() - tailSize;
        buffer.putShort(FREE_SIZE, (short) (upper - HEADER_SIZE));
        buffer.putShort(LOWER, (short) HEADER_SIZE);
        buffer.putShort(UPPER, (short) upper);
        buffer.putShort(TAIL_OFFSET, (short) upper);
        return buffer;
    }

    private void check() {
        if (lower() < HEADER_SIZE) {
            throw new IllegalArgumentException("lower is less than arrayOffset");
        }
        if (lower() > upper()) {
            throw new IllegalArgumentException("lower is greater than upper");
        }
        if (upper() > tailOffset) {
            throw new IllegalArgumentException("upper is greater than itemsOffset");
        }
        if (tailOffset > buffer.limit()) {
            throw new IllegalArgumentException("itemsOffset is out of buffer size");
        }
    }

    public byte version() {
        return buffer.get(VERSION);
    }

    public void version(byte version) {
        buffer.put(VERSION, version);
    }

    public ByteBuffer tail() {
        buffer.position(tailOffset);
        return buffer.slice();
    }

    public void tail(ByteBuffer bytes) {
        if (bytes.limit() != buffer.limit() - tailOffset) {
            throw new IllegalArgumentException("lenght of this bytes array is not equal to tail of page");
        }
        buffer.position(tailOffset);
        buffer.put(bytes);
    }

    public int count() {
        return (lower() - HEADER_SIZE) / ARRAY_CELL_SIZE;
    }

    public boolean hasSpace(byte[] bytes) {
        return hasSpace(bytes.length);
    }

    public boolean hasSpace(ByteBuffer bytes) {
        return hasSpace(bytes.remaining());
    }

    public boolean hasSpace(int itemByteLength) {
        return free() >= itemByteLength + ARRAY_CELL_SIZE;
    }

    public int addItem(byte[] bytes) {
        return addItem(ByteBuffer.wrap(bytes));
    }

    public int addItem(byte[] bytes, int itemIndex) {
        return addItem(ByteBuffer.wrap(bytes), itemIndex);
    }

    public int addItem(ByteBuffer bytes) {
        return addItem(bytes, count());
    }

    public int addItem(ByteBuffer bytes, int itemIndex) {
        bytes.position(0);
        int itemSize = bytes.remaining() + ARRAY_CELL_SIZE;
        if (free() < itemSize) {
            throw new IllegalArgumentException("not enough free space for this item");
        }
        int lower = lower();
        int upper = upper();
        if (upper - lower < itemSize) {
            vakuum();
            lower = lower();
            upper = upper();
        }
        upper = upper - bytes.remaining();

        if (itemIndex < count()) {
            // move upper part of array
            int arrayEntryOffset = arrayEntryOffset(itemIndex);
            byte[] arrayData = new byte[lower - arrayEntryOffset];
            buffer.position(arrayEntryOffset);
            buffer.get(arrayData);
            buffer.position(arrayEntryOffset + ARRAY_CELL_SIZE);
            buffer.put(arrayData);
        }
        // write array cell
        itemOffset(itemIndex, upper);
        itemLength(itemIndex, bytes.remaining());
        // write item data
        buffer.position(upper);
        buffer.put(bytes);
        // adjust the lower and upper values
        lower(lower + ARRAY_CELL_SIZE);
        upper(upper);
        // Important: the new free space is not just upper - lower!
        free(free() - itemSize);
        return itemIndex;
    }

    public ByteBuffer getItem(int index) {
        rangeCheck(index);
        buffer.position(itemOffset(index));
        ByteBuffer result = buffer.slice();
        result.limit(itemLength(index));
        return result;
    }

    public void removeItem(int index) {
        rangeCheck(index);

        int itemLength = itemLength(index);

        // remove array entry
        int arrayEntryOffsetOfNext = arrayEntryOffset(index + 1);
        byte[] arrayData = new byte[lower() - arrayEntryOffsetOfNext];
        buffer.position(arrayEntryOffsetOfNext);
        buffer.get(arrayData);
        buffer.position(arrayEntryOffset(index));
        buffer.put(arrayData);
        lower(lower() - ARRAY_CELL_SIZE);

        // we dont remove the item itself,
        // because it is very expensive

        // Important: the new free space is not just upper - lower!
        free(free() + ARRAY_CELL_SIZE + itemLength);
    }

    public void removeAll() {
        lower(HEADER_SIZE);
        upper(tailOffset);
        buffer.putShort(FREE_SIZE, (short) (tailOffset - HEADER_SIZE));
    }

    public int free() {
        return 0x0000FFFF & buffer.getShort(FREE_SIZE);
    }

    protected void free(int value) {
        buffer.putShort(FREE_SIZE, (short) value);
    }

    /**
     * Defrags the items area.
     */
    public void vakuum() {
        List<byte[]> items = new ArrayList(count());
        for (int index = 0; index < count(); index++) {
            int itemOffset = itemOffset(index);
            int itemLength = itemLength(index);
            byte[] itemData = new byte[itemLength];
            buffer.position(itemOffset);
            buffer.get(itemData);
            items.add(itemData);
        }

        removeAll();

        for(byte[] itemData : items){
            addItem(itemData);
        }
    }

    protected int itemOffset(int index) {
        return 0x0000FFFF & buffer.getShort(arrayEntryOffset(index));
    }

    protected void itemOffset(int index, int value) {
        buffer.putShort(arrayEntryOffset(index), (short) value);
    }

    protected int itemLength(int index) {
        return 0x0000FFFF & buffer.getShort(arrayEntryOffset(index) + 2);
    }

    protected void itemLength(int index, int value) {
        buffer.putShort(arrayEntryOffset(index) + 2, (short) value);
    }

    protected int arrayEntryOffset(int index) {
        return HEADER_SIZE + index * ARRAY_CELL_SIZE;
    }

    protected int lower() {
        return 0x0000FFFF & buffer.getShort(LOWER);
    }

    protected void lower(int value) {
        buffer.putShort(LOWER, (short) value);
    }

    protected int upper() {
        return 0x0000FFFF & buffer.getShort(UPPER);
    }

    protected void upper(int value) {
        buffer.putShort(UPPER, (short) value);
    }

    private void rangeCheck(int index) {
        if (index < 0 || index >= count()) {
            throw new IndexOutOfBoundsException("Index " + index + ", Size " + count());
        }
    }
}
